/*
事件报警数据处理服务

20241122
1.增加网关上线在线离线报警

20240806
1.修复网关的在线离线事件代码错误问题
2.修复event表中设备在线离线事件的deviceid字段错误插入wechat_client问题
3.修复lastvalue表的deviceid字段没有保存网关id，而是保存为仪表id的问题

20240728.1
1.事件表的tag2字段统一为空字符串

2024-07-10
1.订阅[1720589707,1,1]消息，taos保存网关在线离线事件
2.订阅iedgetranser消息，taos保存仪表中断和恢复、遥测越限和恢复、遥信变位事件、遥测要信当前值
*/

package main

import (
	"encoding/json"
	"fmt"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"

	"eventalarm2taos/dataTranser"
	"eventalarm2taos/logger"
	"eventalarm2taos/mqtt"

	aaa "github.com/eclipse/paho.mqtt.golang"
)

// 定义AlarmType的可选值
const (
	Up                 = 201 // 越上限201
	Up2                = 202 // 越上上限202
	Down               = 203 // 越下限203
	Down2              = 204 // 越下下限204
	Recovery           = 205 // 恢复205
	Nochange           = 206 // 呆数据206
	NochangeRecovery   = 207 // 呆数据复归207
	Overturn           = 208 // 翻转208
	OverturnCalcHandle = 209 // 翻转自动处理209
	Intensity          = 210 // 超过烈度210
	IntensityHandle    = 211 // 超过烈度自动抑制211
)

const (
	NOTING       = 100 // 无
	Displacement = 300 // 变位
	//Action = 301 // 动作
	//Return = 302 // 复归

	DevStatus           = 400 // 设备状态
	DevOnline           = 401 // 在线
	DevOffline          = 402 // 离线
	RepeatDevOnline     = 403 // 上线
	GatewayOnline       = 409 // 在线
	GatewayOffline      = 410 // 离线
	RepeatGatewayOnline = 411 // 上线
)

const (
	IOTCOMSRVEX_PUBLISH_TOPIC_PROJECT_ID_INDEX  = 0
	IOTCOMSRVEX_PUBLISH_TOPIC_GATEWAY_ID_INDEX  = 1
	IEDGETRANSER_PUBLISH_TOPIC_PROJECT_ID_INDEX = 0
	IEDGETRANSER_PUBLISH_TOPIC_GATEWAY_ID_INDEX = 2
)

type ConfigJSON struct {
	FileVersion       string `json:"fileVersion"`
	FileLastUpdateUTC int    `json:"fileLastUpdateUTC"`
	ProjectID         string `json:"projectId"`
	BrokerIP          string `json:"brokerIp"`
	BrokerPort        int    `json:"brokerPort"`
	BrokerUser        string `json:"brokerUser"`
	BrokerPWD         string `json:"brokerPWD"`
	BrokerCa          string `json:"brokerCa"`
	MqttClientID      string `json:"mqttClientID"`
	MqttSubscribes    string `json:"mqttSubscribes"`
	TaosIP            string `json:"taosIp"`
	TaosPort          int    `json:"taosPort"`
	TaosUser          string `json:"taosUser"`
	TaosPWD           string `json:"taosPWD"`
}

type DevModelVal struct {
	DevList []struct {
		DeviceID string `json:"id"`
		T        int    `json:"t"`
		D        []struct {
			Pn string `json:"pn"`
			V  string `json:"v"`
		} `json:"d"`
	} `json:"devList"`
}

type GatewayInfo struct {
	GatewayID  string    `json:"id"`
	ProjectID  string    `json:"project_id"`
	LastTime   time.Time `json:"last_time"`
	IsOffline  bool      `json:"is_offline"`
	OffLineCnt int       `json:"offline_cnt"`
}

var mqttList []mqtt.MqttStru

func main() {

	// 关闭 logger 和 logFile
	defer func() {
		if r := recover(); r != nil {
			logger.Error(fmt.Sprintf("main panic : %v", r))
		}
		logger.GetLogger().Sync()
		//程序结束
	}()

	const procName = "eventalarm2taos"
	const version = "v20241122"

	// 使用 logger 输出日志
	logger.InitLogger(procName)
	logger.Info(fmt.Sprintf("%s %s", procName, version))

	//读取配置
	logger.Info("load config")
	configFile := fmt.Sprintf("conf/%s.json", procName)
	fileContent, err := os.ReadFile(configFile)
	if err != nil {
		logger.Error(fmt.Sprintf("%v", err))
		return
	}

	// 使用 json.Unmarshal 进行 JSON 反序列化
	var configData ConfigJSON
	err = json.Unmarshal(fileContent, &configData)
	if err != nil {
		logger.Error(fmt.Sprintf("%v", err))
		return
	}

	//初始化taos连接
	logger.Info("init taos")
	taos := &dataTranser.HttpTaos{}
	err = taos.Init(configData.TaosIP, configData.TaosPort, configData.TaosUser, configData.TaosPWD)
	if err != nil {
		logger.Error(fmt.Sprintf("%v", err))
		return
	}

	//初始化mqtt发送
	mqttList = make([]mqtt.MqttStru, 0)
	mqttList = append(mqttList, mqtt.MqttStru{
		ClientID:  configData.MqttClientID,
		Subscribe: strings.Split(configData.MqttSubscribes, "&&"),
		LastRece:  time.Now(),
	})
	mqttList[0].MsgChannel = make(chan mqtt.MsgData, 1000)
	err = mqttList[0].Init(
		configData.MqttSubscribes,
		false,
		configData.BrokerUser,
		configData.BrokerPWD,
		configData.BrokerIP,
		configData.BrokerPort,
		configData.BrokerCa,
		onConnectionLost,
		onConnect)
	if err != nil {
		logger.Error(fmt.Sprintf("%v", err))
		return
	}

	gatewayList := map[string]GatewayInfo{}

	ticker30Sec := time.NewTicker(30 * time.Second)
	time5Min := time.Now()
	insertSuccCnt := 0
	insertFailCnt := 0
	msgCnt := 0
	for {
		select {
		case <-ticker30Sec.C:

			for _, gateway := range gatewayList {
				if time.Since(gateway.LastTime).Seconds() > float64(60*(5+gateway.OffLineCnt)) {

					err = taos.SaveEvent(gateway.GatewayID, "", gateway.ProjectID, GatewayOffline, gateway.GatewayID, "offline")
					if err != nil {
						insertFailCnt++
						logger.Error(fmt.Sprintf("save event fail : %v", err))
					} else {
						insertSuccCnt++
					}

					gatewayList[gateway.GatewayID] = GatewayInfo{
						GatewayID:  gateway.GatewayID,
						ProjectID:  gateway.ProjectID,
						LastTime:   time.Now(),
						IsOffline:  true,
						OffLineCnt: gateway.OffLineCnt + 1,
					}
				}
			}

			logger.Debug("main loop")
		default:
			for i := range mqttList {
				if len(mqttList[i].MsgChannel) > 0 {
					msgCnt++

					mqttMsg := <-mqttList[i].MsgChannel
					logger.Debug(fmt.Sprintf("mqtt %s rece from %s msg %s", mqttList[i].ClientID, strings.Join(mqttMsg.Topic, "/"), string(mqttMsg.Payload)))

					mqttList[i].LastRece = time.Now()

					msg := string(mqttMsg.Payload)

					if msg[0] == '[' && msg[len(msg)-1] == ']' {
						macAddrPattern := regexp.MustCompile(`^([0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}$`)
						if !macAddrPattern.MatchString(strings.Split(msg[1:len(msg)-1], ",")[0]) {
							err := taos.SaveHart(mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_GATEWAY_ID_INDEX], mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_PROJECT_ID_INDEX], msg)
							if err != nil {
								insertFailCnt++
								logger.Error(fmt.Sprintf("save hart fail : %v", err))
							} else {
								insertSuccCnt++
							}

							if gateway, ok := gatewayList[mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_GATEWAY_ID_INDEX]]; !ok {
								err = taos.SaveEvent(mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_GATEWAY_ID_INDEX], "", mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_PROJECT_ID_INDEX], GatewayOnline, mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_GATEWAY_ID_INDEX], "online")
							} else if !gateway.IsOffline {
								err = taos.SaveEvent(mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_GATEWAY_ID_INDEX], "", mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_PROJECT_ID_INDEX], GatewayOnline, mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_GATEWAY_ID_INDEX], "online")
							} else {
								err = taos.SaveEvent(mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_GATEWAY_ID_INDEX], "", mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_PROJECT_ID_INDEX], RepeatGatewayOnline, mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_GATEWAY_ID_INDEX], "online")
							}
							if err != nil {
								insertFailCnt++
								logger.Error(fmt.Sprintf("save event fail : %v", err))
							} else {
								insertSuccCnt++
							}

							// 更新网关心跳时间
							gatewayList[mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_GATEWAY_ID_INDEX]] = GatewayInfo{
								GatewayID:  mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_GATEWAY_ID_INDEX],
								ProjectID:  mqttMsg.Topic[IOTCOMSRVEX_PUBLISH_TOPIC_PROJECT_ID_INDEX],
								LastTime:   time.Now(),
								IsOffline:  false,
								OffLineCnt: 0,
							}

						} else {
							logger.Warn(fmt.Sprintf("give up hart data : %s", msg))
						}
					} else {
						var jsonData map[string]interface{}
						err = json.Unmarshal(mqttMsg.Payload, &jsonData)
						if err != nil {
							logger.Error(fmt.Sprintf("unmarshal mqtt msg to json fail : %v", err))
						} else {
							if _, ok := jsonData["KeyPart1"]; ok {
								logger.Warn("unsupported keypart1,2,3 process")

							} else if _, ok := jsonData["devList"]; ok {
								var devModel DevModelVal
								err = json.Unmarshal(mqttMsg.Payload, &devModel)
								if err == nil {
									for _, dev := range devModel.DevList {
										for _, pns := range dev.D {
											if pns.Pn == fmt.Sprintf("%s_err", dev.DeviceID) {
												eventCode := DevOnline
												eventDesc := "online"
												if pns.V == "1" {
													eventCode = DevOffline
													eventDesc = "offline"
												}
												err = taos.SaveEvent(mqttMsg.Topic[IEDGETRANSER_PUBLISH_TOPIC_GATEWAY_ID_INDEX], "", mqttMsg.Topic[IEDGETRANSER_PUBLISH_TOPIC_PROJECT_ID_INDEX], eventCode, dev.DeviceID, eventDesc)
												if err != nil {
													insertFailCnt++
													logger.Error(fmt.Sprintf("save %s-%s %s event fail : %v", mqttMsg.Topic[IEDGETRANSER_PUBLISH_TOPIC_GATEWAY_ID_INDEX], dev.DeviceID, eventDesc, err))
												} else {
													insertSuccCnt++
												}
											}

											val, err := strconv.ParseFloat(pns.V, 64)
											if err != nil {
												insertFailCnt++
												logger.Error(fmt.Sprintf("save %s-%s %s data fail : %v", mqttMsg.Topic[IEDGETRANSER_PUBLISH_TOPIC_GATEWAY_ID_INDEX], dev.DeviceID, pns.Pn, err))
											} else {
												err = taos.SaveData(pns.Pn, mqttMsg.Topic[IEDGETRANSER_PUBLISH_TOPIC_GATEWAY_ID_INDEX], mqttMsg.Topic[IEDGETRANSER_PUBLISH_TOPIC_PROJECT_ID_INDEX], val)
												if err != nil {
													insertFailCnt++
													logger.Error(fmt.Sprintf("save %s-%s %s data fail : %v", mqttMsg.Topic[IEDGETRANSER_PUBLISH_TOPIC_GATEWAY_ID_INDEX], dev.DeviceID, pns.Pn, err))
												} else {
													insertSuccCnt++
												}
											}
										}
									}
								} else {
									logger.Error(fmt.Sprintf("unmarshal devModel fail : %v", err))
								}
							} else {
								logger.Error("unknown mqtt msg")
							}
						}
					}

				} else if time.Since(mqttList[i].LastRece).Seconds() > 60 {
					mqttList[i].Close()
					token := mqttList[i].Client.Connect()
					token.Wait()
					if token.Error() == nil {
						mqttList[i].LastRece = time.Now()
						mqttList[i].ReConnectCnt = 0
					} else {
						logger.Error(fmt.Sprintf("mqtt %s reconnect fail : %v", mqttList[i].ClientID, token.Error()))
						mqttList[i].LastRece = time.Now().Add(-30 * time.Second)
						mqttList[i].ReConnectCnt++
						if mqttList[i].ReConnectCnt >= 5 {
							panic(fmt.Sprintf("mqtt %s 30min none data!", mqttList[i].ClientID))
						}
					}
				}
			}
		}

		if time.Since(time5Min).Seconds() > 300 &&
			time.Now().Minute()%5 == 0 &&
			time.Now().Second() < 30 {
			time5Min = time.Now()

			leftMsgCnt := 0
			for i := range mqttList {
				leftMsgCnt += len(mqttList[i].MsgChannel)
			}

			logger.Info(fmt.Sprintf("5min total : insertSuccCnt:%d, insertFailCnt:%d, msgCnt:%d, leftMsgCnt:%d", insertSuccCnt, insertFailCnt, msgCnt, leftMsgCnt))
			insertSuccCnt = 0
			insertFailCnt = 0
			msgCnt = 0
		}

		time.Sleep(100 * time.Millisecond)
	}
}

// onConnectionLost 是 MQTT 客户端连接丢失时的回调函数
// 参数：
//
//	client mqtt.Client：MQTT 客户端对象
//	err error：连接丢失的错误信息
//
// 返回值：无
func onConnectionLost(client aaa.Client, err error) {
	opts := client.OptionsReader()
	logger.Error(fmt.Sprintf("mqtt %s connection break from %s, reason : %v", opts.ClientID(), opts.Servers(), err))
}

// onConnect 是一个处理MQTT客户端连接事件的函数
// 当MQTT客户端成功连接时，该函数将被调用
// 参数：
//
//	client aaa.Client：MQTT客户端对象
func onConnect(client aaa.Client) {
	opts := client.OptionsReader()
	logger.Info("mqtt %s connected to %s", opts.ClientID(), opts.Servers())

	for _, item := range mqttList {
		if item.ClientID == opts.ClientID() {
			logger.Info(fmt.Sprintf("start mqtt %s receive and subscribe %v!", item.ClientID, item.Subscribe))
			err := mqttList[0].Recv()
			if err != nil {
				logger.Error(fmt.Sprintf("%v", err))
			}
		}
	}
}
